package com.unism.uniMeterage.meterage.stream;

import com.unism.uniMeterage.meterage.stream.version1_1.DefaultMessage;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class AbstractWorker extends Thread implements Worker {

   private static final Integer DefaultMaxMessageNum=100;//默认worker缓存最大消息数
   private Integer maxMessageNum;
   private AtomicInteger status;
   private AtomicInteger currNum;
   private String workerName;
   private ConcurrentLinkedQueue<Message> queue;
   private Manager manager;

    /**
     * 初始化Worker本身，资源、状态的初始化
     * @param workerName
     * @param manager
     * @param group
     * @param maxMessageNum
     */
    public void initWorker(String workerName, Manager manager, ThreadGroup group, Integer maxMessageNum) {
        this.setDaemon(true);
        this.setPriority(Math.max(Thread.MIN_PRIORITY, Thread.currentThread().getPriority() - 1));

        this.workerName = "DefaultWorker==>"+workerName;
        this.manager=manager;
        if(maxMessageNum == null || maxMessageNum <=0) {
            this.maxMessageNum=DefaultMaxMessageNum;
        }else{
            this.maxMessageNum=maxMessageNum;
        }
        this.queue=new ConcurrentLinkedQueue<Message>();
        this.status=new AtomicInteger(WorkerStatus.NEW.getCode());
        this.currNum=new AtomicInteger(0);
        //供业务初始化保留接口
        this.init();
    }
    @Override
    public void init() {
       System.out.println("worker:"+this.getWorkerName()+"init");
    }
    @Override
    public Boolean execute(Message data) {
        Long num=Long.valueOf(data.getData().toString());
        num++;
        System.out.println(data.getData()+"===="+System.currentTimeMillis());
        this.manager.sendMessage(new DefaultMessage(this,"liuc",num),this,this);
        return true;
    }
    /**
     * 激活当前Worker,一个Worker只能调用一次
     */
    @Override
    public void activate() {
        if(this.getWorkerName() == null || "".equals(this.getWorkerName())){
            throw new IllegalStateException("workerName is invalid");
        }
        if(queue == null ){
            throw new IllegalStateException("queue is invalid");
        }
        if(manager == null ){
            throw new IllegalStateException("manager is invalid");
        }
        if(this.getMessageCount() >= this.getMaxMessageCount()){
            throw new IllegalStateException("too many message in the queue");
        }
        if(this.currStatus().getCode().intValue() == WorkerStatus.DESTORY.getCode().intValue()){
            throw new IllegalStateException("the worker already was destoried");
        }
        //将worker状态设置为WAITING
        this.setStatus(WorkerStatus.WAITING);
    }
    /**
     * 此方法将强制销毁Worker
     */
    @Override
    public void destory() {
     //将worker状态设置为Destory
      status.getAndSet(WorkerStatus.DESTORY.getCode());
      //清除缓存中的内容
      this.queue.clear();
      //销毁缓存
      this.queue=null;
      //销毁外部引用
      this.manager=null;
    }
    @Override
    public String getWorkerName() {
        return this.workerName;
    }

    @Override
    public WorkerStatus  currStatus() {
        return WorkerStatus.getStatusByCode(status.get());
    }
    @Override
    public void setStatus(WorkerStatus status) {
        this.status.getAndSet(status.getCode());
    }
    @Override
    public Integer getMaxMessageCount() {
        return this.maxMessageNum;
    }

    /**
     * 此方面尽量少用，原因有2
     * 1、此方法为保证数据可靠采用了排它锁，而不是乐观锁
     * 2、此方面是默认遍历ConcurrentLinkedQueue所有节点
     * 3、后期考虑自己做缓存优化
     * @return
     */
    @Override
    public Integer getMessageCount() {
        synchronized (queue) {
            return this.queue.size();
        }
    }
    /**
     * @param aa 无效，单纯多态
     * @return
     */
    public Integer getMessageCount(int aa) {
        return this.currNum.get();
    }
    @Override
    public Message doNext() {
        if(!this.queue.isEmpty()){
            Message message=this.queue.poll();
            if(message !=null){
                this.currNum.getAndDecrement();
                return message;
            }
        }
        return null;
    }
    @Override
    public Boolean addMessage(Message message) {
        if(this.queue !=null ){
            int currNum=this.currNum.get();
            if(this.getMessageCount(0) < this.getMaxMessageCount()){
                if(this.status.get() != WorkerStatus.DESTORY.getCode().intValue()){
                    this.currNum.incrementAndGet();
                    this.queue.offer(message);
                    return true;
                }
            }
        }
        return false;
    }
    @Override
    public void run(){
        System.out.println("thread run.....");
       while ( !Thread.currentThread().isInterrupted()
                && status.intValue() != WorkerStatus.DESTORY.getCode().intValue()) {
            try {
                DefaultMessage message=(DefaultMessage)doNext();
                if(message !=null){
                    setStatus(WorkerStatus.WORKING);
                    try {
                        message.triggerListeners(new MessageEvent(this, message, MessageEvent.MessageStatus.DELIVERED));
                        execute(message);
                        message.triggerListeners(new MessageEvent(this, message, MessageEvent.MessageStatus.COMPLETED));
                    } catch (Exception e) {
                        message.triggerListeners(new MessageEvent(this, message, MessageEvent.MessageStatus.FAILED));
                        System.out.println("execute 执行过程中出现错误");
                    }

                }else{
                    synchronized ( this.queue) {
                        setStatus(WorkerStatus.WAITING);
                        this.queue.wait(100);
                    }
                }
            }catch (InterruptedException e) {
                Thread.currentThread().interrupt();//重新设置中断标示
            }
        }
    }
    public void notifyWorker(){
        this.queue.notifyAll();
    }
    public static enum WorkerStatus {
        NEW("初始化",1),
        WORKING("工作中",2),
        WAITING("等待中",3),
        DESTORY("销毁中",4);
        private String name;
        private Integer code;
        public String getName() {
            return name;
        }
        public void setName(String name) {
            this.name = name;
        }
        public Integer getCode() {
            return code;
        }
        public void setCode(Integer code) {
            this.code = code;
        }
        WorkerStatus(String name, Integer code) {
            this.name = name;
            this.code = code;
        }
        public static WorkerStatus getStatusByCode(Integer code){
            for (WorkerStatus s : WorkerStatus.values()){
                if(s.getCode().intValue() == code){
                    return s;
                }
            }
            return  null;
        };
    }
}
